package edu.uab.enronemail.mapreduce;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.james.mime4j.stream.MimeConfig;
import org.apache.james.mime4j.stream.MimeTokenStream;

import edu.uab.enronemail.factory.ElasticSearchClientFactory;
import edu.uab.enronemail.helper.SequenceFileReader;
import edu.uab.enronemail.request.DocumentLoader;
import edu.uab.enronemail.request.EmailDocument;

public class IndexCreatorMapper extends Mapper<Text, Text, NullWritable, NullWritable> {
	
	private static MimeConfig mimeConfig = new MimeConfig();
	private static MimeTokenStream mimeStream = null;
	DocumentLoader documentLoader = null;
	BufferedReader fileReader = null;

	@Override
	public void setup(Context context) {
		System.out.println("Beginning configuring");
		mimeConfig.setMaxHeaderLen(-1);
		mimeConfig.setMaxLineLen(-1);
		mimeStream = new MimeTokenStream(mimeConfig);
		documentLoader = DocumentLoader
				.getDocumentLoader(ElasticSearchClientFactory.getClient());

		Path[] paths = null;
		try {
			paths = DistributedCache.getLocalCacheFiles(context
					.getConfiguration());
			for (Path path : paths) {
				if (path.toString().contains("stopwords")) {
					fileReader = new BufferedReader(new FileReader(new File(
							path.toString())));
					break;
				}
			}

		} catch (IOException e) {
			e.printStackTrace();
			new RuntimeException(e);
		}

		EmailDocument.configureStopWords(fileReader);
	}

	@Override
	public void cleanup(Context context) throws IOException {
		documentLoader.completeBulkExecution();
	}

	@Override
	public void map(Text key, Text value, Context context) throws IOException {
		EmailDocument emailDoc = SequenceFileReader.parseEmail(
				value.toString(), mimeStream);
		documentLoader.addDocument(emailDoc);
	}

}
